package com.tiantang.kafka.producer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * 
 * @description:功能描述 添加了拦截器的生产者
 * @author: liujinkun
 * @date 2018年5月27日
 * @version:1.0
 */
public class MyProducerWithInterceptor {

	public void produce() {
		// 1.配置生产者的属性
		Map<String, Object> configs = new HashMap<String, Object>();

		// kafka集群地址,可以是多个(注意：这里是Kafka的集群，端口号9092是Kafka的默认端口号，不要写成zookeeper的端口号)
		configs.put("bootstrap.servers", "com.tiantang.hadoop01:9092");

		// 发送的消息是否等待应答(-1和all表示leader和folowers确认，0表示不经过任何确认，1表示只经过leader确认)
		configs.put("acks", "all");

		// 消息发送失败的重试次数
		configs.put("retries", 0);

		// 批量处理数据的大小，单位：字节,默认16KB
		configs.put("batch.size", 16384);

		// 批量处理数据的延迟，单位：ms
		configs.put("linger.ms", 5);

		// 设置内存缓冲区,默认32M
		configs.put("buffer.memory", 33554432);

		// 设置key的序列化类
		configs.put("key.serializer", org.apache.kafka.common.serialization.StringSerializer.class);

		// 设置value的序列化类
		configs.put("value.serializer", org.apache.kafka.common.serialization.StringSerializer.class);
		
		// 设置拦截器链
		List<String> interceptors=new ArrayList<String>();
		interceptors.add("com.tiantang.kafka.interceptor.TimeInterceptor");
		interceptors.add("com.tiantang.kafka.interceptor.CountInterceptor");
		configs.put("interceptor.classes", interceptors);

		// 2.实例化Kafka生产者
		KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);

		// 3.生产消息,调用producer的send()方法发送消息
		for (int i = 0; i < 50; i++) {
			// 采用默认分区类 DefaultPartitioner
			// 不指定key则采用轮询的方式计算分区，指定key则采用hash key的计算方式计算分区
			producer.send(new ProducerRecord<String, String>("test", "record" + i));
		}

		// 4.关闭连接
		producer.close();
	}
}
